package com.message.queue.rocket.translation;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * Created by 张波 on 2019/10/19.
 */
public class TransactionListenerImpl implements TransactionListener {
    /**
     * 异步消息处理方法
     *
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        String args = (String) arg;
        System.out.println("执行本地事物args:" + args + ",message:" + msg);
        //TODO 此处进行实际要执行的业务操作

        //COMMIT_MESSAGE:提交消息（执行成功返回，只有commit状态的才会被消费端发现），ROLLBACK_MESSAGE：回滚消息（执行失败返回），UNKNOW：消息不可达(让MQ自已来调用方法回查状态，在状态不可知时返回)
//        return LocalTransactionState.COMMIT_MESSAGE;
        return LocalTransactionState.UNKNOW;
    }

    /**
     * MQ定时回查未确认消息方法
     *
     * @param msg
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("回调消息检查返回keys:" + msg.getKeys());
        String keys = msg.getKeys();
        if (keys.contains("1") ||
                keys.contains("3") ||
                keys.contains("5") ||
                keys.contains("7") ||
                keys.contains("9")) {
            //当检查状态为失败时，返回ROLLBACK_MESSAGE
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        if (keys.contains("8")) {
            //状态不可知时返回
            return LocalTransactionState.UNKNOW;
        }
        //当查询到状态为执行成功时
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}
